package com.xkh.rocketmq.service;

import com.xkh.rocketmq.config.JmsConfig;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.util.List;

@Configuration
public class RocketMQAutoConfiguration {

    @Resource
    private ApplicationEventPublisher publisher;

    /**
     * 初始化向rocketMQ发送普通消息的生产着
     * ProducerGroup这个概念发送普通的消息时，作用不大，但是发送分布式事务消息时，比较关键，
     * 因为服务器会回查这个Group下的任意一个Producer
     */
    @Bean
    public DefaultMQProducer defaultMQProducer() {
        //生产者的组名
        DefaultMQProducer producer = new DefaultMQProducer(JmsConfig.PRODUCER_GROUP_NAME);
        //指定NameServer地址，多个地址以 ; 隔开
        producer.setNamesrvAddr(JmsConfig.NAME_SERVER);
        //不开启vip通道 开通端口会减2
        producer.setVipChannelEnabled(false);
        //消息发送失败重试次数
        producer.setRetryTimesWhenSendFailed(3);
        //异步发送失败重试次数
        producer.setRetryTimesWhenSendAsyncFailed(3);
        //消息没有发送成功，是否发送到另外一个Broker中
        producer.setRetryAnotherBrokerWhenNotStoreOK(true);
        try {
            producer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return producer;
    }


    /**
     * 初始化rocketMQ消息监听方式的消费者
     * MessageModel
     * CLUSTERING 集群模式(默认)
     * BROADCASTING 广播模式
     */
    @Bean
    public void pushConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(JmsConfig.CONSUMER_GROUP_NAME);
        consumer.setNamesrvAddr(JmsConfig.NAME_SERVER);
        //CONSUME_FROM_FIRST_OFFSET：第一次启动从队列初始位置消费，后续再启动接着上次消费的进度开始消费
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setMessageModel(MessageModel.CLUSTERING);
        // 订阅topic，可以对指定消息进行过滤
        List<String> subscribeList = JmsConfig.subscribe;
        for (String subscribe : subscribeList) {
            consumer.subscribe(subscribe.split(":")[0], subscribe.split(":")[1]);
        }
//        consumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {
//            this.publisher.publishEvent(new RocketMQEvent(list, consumer));
//            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//        });
        consumer.registerMessageListener((MessageListenerOrderly) (list, consumeConcurrentlyContext) -> {
            this.publisher.publishEvent(new RocketMQEvent(list, consumer));
            return ConsumeOrderlyStatus.SUCCESS;
        });
        consumer.start();
    }
}
